package com.bbmall.dwd;

import com.bbmall.SparkSessionUtil;
import org.apache.spark.sql.SparkSession;

/**
 * 会员消费事实表抽取到DWD层
 */


public class OdsToDwdFactMemberConsumption {
    public static void main(String[] args) {
        // 解析参数
        if (args.length < 2) {
            System.err.println("参数错误！用法：");
            System.err.println("历史全量处理：<is_full=true> <任意值>（历史数据在99991231分区）");
            System.err.println("每日增量处理：<is_full=false> <biz_date=yyyy-MM-dd>（如：false 2025-10-18）");
            System.exit(1);
        }
        boolean isFull = Boolean.parseBoolean(args[0]);
        String bizDate = args[1];

        // 动态生成ODS分区条件
        String odsWhereClause;
        if (isFull) {
            odsWhereClause = "dt = '99991231'";
        } else {
            String odsDt = bizDate.replace("-", "");
            odsWhereClause = "dt = '" + odsDt + "'";
        }

        // 核心SQL：严格匹配dwd_fact_member_consumption表结构
        String sql = (isFull ? "INSERT OVERWRITE" : "INSERT INTO") + " TABLE dwd.dwd_fact_member_consumption PARTITION (consumption_date) " +
                "SELECT " +
                // 1. 基础ID字段（包含store_code、transaction_id）
                "record_id, " +                                  // record_id (BIGINT)
                "member_id, " +                                  // member_id (STRING)
                "store_code, " +                                 // store_code (STRING)
                "transaction_id, " +                             // transaction_id (BIGINT)
                // 2. 时间维度字段（包含consumption_quarter）
                "consumption_datetime, " +                       // consumption_datetime (STRING)
                "date_format(consumption_datetime, 'yyyy-MM-dd') AS consumption_date, " + // consumption_date (分区字段)
                "year(consumption_datetime) AS consumption_year, " + // consumption_year (INT)
                "month(consumption_datetime) AS consumption_month, " + // consumption_month (INT)
                "quarter(consumption_datetime) AS consumption_quarter, " + // consumption_quarter (INT)
                "dayofweek(consumption_datetime) AS day_of_week, " + // day_of_week (INT)
                "CASE WHEN dayofweek(consumption_datetime) IN (1,7) THEN 1 ELSE 0 END AS is_weekend, " + // is_weekend (TINYINT)
                // 3. 度量字段（与DWD表完全一致）
                "consumption_amount, " +                         // consumption_amount (DECIMAL)
                "earned_points, " +                              // earned_points (INT)
                // 4. 创建时间（STRING类型）
                "created_time AS create_time " +                 // create_time (STRING)
                // 数据源
                "FROM ods.ods_member_consumption_record " +
                "WHERE " + odsWhereClause + ";";

        // 执行
        SparkSession spark = SparkSessionUtil.getSparkSession("ODS_TO_DWD_FACT_MEMBER_CONSUMPTION");
        try {
            spark.sql(sql);
            System.out.println("会员消费事实表DWD层" + (isFull ? "历史全量" : "增量（" + bizDate + "）") + "处理完成！");
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            SparkSessionUtil.close();
        }
    }
}